package site.ruyi.devclient.client.zookeeper.impl;

import java.io.IOException;
import java.io.OutputStream;
import java.io.PrintStream;
import java.util.List;
import java.util.Map;
import java.util.Optional;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.cli.CliException;
import org.apache.zookeeper.data.ACL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javafx.scene.control.TextArea;
import site.ruyi.devclient.client.zookeeper.IZookeeperClient;
import site.ruyi.devclient.core.AlertUtil;
import site.ruyi.devclient.core.FormDialog;
import site.ruyi.devclient.core.UserData;

/**
 * 单台Zookeeper连接客户端
 * @author self
 *
 */
public class ZookeeperClientImpl implements IZookeeperClient,Watcher {

	private static final Logger log=LoggerFactory.getLogger(ZookeeperClientImpl.class);

	//客户端名称
	private static final String CLIENT_NAME = "Zookeeper客户端";

	//Jedis客户端
	private ZooKeeper zk;

	private ZookeeperCmd zkCmd;

	private ConnectInfo connectInfo;

	public ZookeeperClientImpl(){
		UserData.getDB().exec("CREATE TABLE IF NOT EXISTS zookeeper_client(id TEXT NOT NULL ON CONFLICT FAIL UNIQUE ON CONFLICT REPLACE,servers TEXT,create_time TIMESTAMP default (datetime('now', 'localtime')))");
	}

	@Override
	public String getName() {
		return CLIENT_NAME;
	}

	@Override
	public ConnectInfo getConnectInfo() {
		return connectInfo;
	}

	@Override
	public boolean init(ZKOutputListener outputListener) {
		log.info("初始化客户端组件！");
		return createConnection(outputListener);
	}

	/**
	 * 创建连接
	 * @return
	 */
	private boolean createConnection(ZKOutputListener outputListener) {
		log.info("创建Socket连接！");
		TextArea serverTextArea = new TextArea("127.0.0.1:2181");

		Map<String,Object> lastConnInfo=UserData.getDB().find("select servers from zookeeper_client order by create_time desc limit 1");
		if(lastConnInfo!=null&&!lastConnInfo.isEmpty()){
			serverTextArea.setText(String.valueOf(lastConnInfo.get("servers")));
		}

		FormDialog newConnForm = new FormDialog("新建链接"
				,new FormDialog.Item("主机：", serverTextArea));
		Optional<Boolean> result = newConnForm.showAndWait();
		if(result.isPresent()) {
			String services = serverTextArea.getText();
			log.info("Socket连接参数 -> host:"+services);
			this.connectInfo = new ConnectInfo(services);
			try {
				zk = new ZooKeeper(services, 5000, this);
				zkCmd = new ZookeeperCmd(zk,new CmdPrintStream(System.out, outputListener));
				outputListener.handle("已连接！\n");
				log.info("已连接到Zookeeper服务器。");

				String connId=DigestUtils.md5Hex(services);
				UserData.getDB().save("insert into zookeeper_client(id,servers) values(?,?)", connId,services);

				return true;
			} catch (IOException e) {
				AlertUtil.exception("连接异常", e);
			}
		}
		return false;
	}

	@Override
	public String exeCmd(String cmd)  {
		if(cmd==null||"".equals(cmd.trim())) {
			return null;
		}
		log.info("执行命令："+cmd);
		//命令如果没有换行符则补充一个
		if(!cmd.endsWith("\n")) {
			cmd+="\n";
		}

		try {
			zkCmd.executeLine(cmd);
		} catch (CliException | InterruptedException | IOException e) {
			e.printStackTrace();
		}

		return null;
	}

	@Override
	public void close() throws IOException {
		if(zk!=null) {
			try {
				zk.close();
			} catch (InterruptedException e) {
				AlertUtil.exception(e);
			}
		}
	}

	@Override
	public void process(WatchedEvent event) {
		log.info("Watch received event: "+event);
	}

	/**
	 * 自定义命令输出组件
	 * @author self
	 *
	 */
	private static class CmdPrintStream extends PrintStream{

		private ZKOutputListener outputListener;

		public CmdPrintStream(OutputStream out,ZKOutputListener outputListener) {
			super(out);
			this.outputListener = outputListener;
		}

		@Override
		public void write(byte[] buf, int off, int len) {
			final String message = new String(buf,off,len);
			outputListener.handle(message);
		}
	}

	@Override
	public void create(String path, CreateMode mode) {
		try {
			zk.create(path, null, Ids.OPEN_ACL_UNSAFE, mode);
		} catch (KeeperException | InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public void delete(String path) {
		try {
			zk.delete(path, -1);
		} catch (InterruptedException | KeeperException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public List<String> getChildren(String path) {
		try {
			return zk.getChildren(path, false);
		} catch (KeeperException | InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public List<ACL> getACL(String path) {
		try {
			return zk.getACL(path, null);
		} catch (KeeperException | InterruptedException e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public byte[] getData(String path) {
		try {
			return zk.getData(path, false, null);
		} catch (KeeperException | InterruptedException e) {
			throw new RuntimeException(e);
		}
	}
}
